package com.sdg.dataload

import java.net.InetAddress

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions.{collect_set, concat_ws}
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import org.elasticsearch.spark.sql.EsSparkSQL
import org.elasticsearch.transport.client.PreBuiltTransportClient

/**
  * 数据初始化
  */
object DataLoader {
  val ES_HOST_PORT_REGEX = "(.+):(\\d+)".r
  //mongo url
  val MONGO_URI: String = "mongodb://vm2:27017/recom3"
  //mongo db
  val MONGODB_DATABASE: String = "recom3"
  //9200 es和外部通信的端口
  var ES_HTTPHOSTS = "vm2:9200"
  //9300 es节点之间同行的端口
  var ES_TRANSPORTHOSTS = "vm2:9300"

  //es 中的索引名称
  var ES_INDEX = "recom3"
  //es集群的名称
  var ES_CLUSTER_NAME = "my-es"

  //电影数据表名
  var MONGODB_MOVIE_COLLECTION = "Movie"

  //电影评分表名 [ ]
  var MONGODB_RATING_COLLECTION = "Rating"
  //电影标签的表名
  var MONGODB_TAG_COLLECTION = "Tag"

  //mongo的驱动
  val MONGO_DRVIVE_CLASS = "com.mongodb.spark.sql"
  //es的驱动
  val ES_DRVIVE_CLASS = "com.elasticsearch.spark.sql"

  //ES 中的电影数据集
  var ES_MOVIE_TYPE = "Movie"

  /**
    * 将数据写入到mongodb
    *
    * @param movieDF
    * @param ratingDF
    * @param tagDF
    */
  def storeDataInMongo(movieDF: DataFrame, ratingDF: DataFrame, tagDF: DataFrame)(implicit mongoConf: MongoConfig) = {
    // 创建到MongoDB的连接
    val mongoClient = MongoClient(MongoClientURI(mongoConf.uri))
    //这里可以优化一下  做个判断  有则删除   无则插入

    // 删除Movie的Collection
    mongoClient(mongoConf.db)(MONGODB_MOVIE_COLLECTION).dropCollection()
    // 删除Rating的Collection
    mongoClient(mongoConf.db)(MONGODB_RATING_COLLECTION).dropCollection()
    // 删除Tag的Collection
    mongoClient(mongoConf.db)(MONGODB_TAG_COLLECTION).dropCollection()
    //将Movie数据集写入到MongoDB
    movieDF
      .write
      .option("uri", mongoConf.uri)
      .option("collection", MONGODB_MOVIE_COLLECTION)
      .mode("overwrite")
      .format(MONGO_DRVIVE_CLASS)
      .save()
    //将Rating数据集写入到MongoDB
    ratingDF
      .write.option("uri", mongoConf.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .mode("overwrite")
      .format(MONGO_DRVIVE_CLASS)
      .save()
    //将Tag数据集写入到MongoDB
    tagDF
      .write.option("uri", mongoConf.uri)
      .option("collection", MONGODB_TAG_COLLECTION)
      .mode("overwrite")
      .format(MONGO_DRVIVE_CLASS)
      .save()
    //创建索引  1正序  2倒序
    mongoClient(mongoConf.db)(MONGODB_MOVIE_COLLECTION).createIndex(MongoDBObject("mid" -> 1))
    mongoClient(mongoConf.db)(MONGODB_RATING_COLLECTION).createIndex(MongoDBObject("mid" -> 1))
    mongoClient(mongoConf.db)(MONGODB_RATING_COLLECTION).createIndex(MongoDBObject("uid" -> 1))
    mongoClient(mongoConf.db)(MONGODB_TAG_COLLECTION).createIndex(MongoDBObject("mid" -> 1))
    mongoClient(mongoConf.db)(MONGODB_TAG_COLLECTION).createIndex(MongoDBObject("uid" -> 1))
    //关闭MongoDB的连接
    mongoClient.close()

  }


  /**
    * 将数据存储在es中
    *
    * @param esMovieDF
    */
  def storeDataInES(esMovieDF: DataFrame)(implicit esConf: ESConfig) = {
    // 需要操作的Index名称
    val indexName = esConf.index
    // 新建一个到ES的连接配置
    var settings: Settings = Settings.builder().put("cluster.name", esConf.clusterName).build()
    // 创建到ES的连接客户端
    val esClient = new PreBuiltTransportClient(settings)

    //对于设定的多个Node分别通过正则表达式进行模式匹配，并添加到客户端实例
    esConf.transportHosts.split(";")
      .foreach {
        case ES_HOST_PORT_REGEX(host: String, port: String) =>
          esClient.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port.toInt))
      }
    // 检查如果Index存在，那么删除Index
    if (esClient.admin().indices().exists(new IndicesExistsRequest(indexName)).actionGet().isExists) {
      // 删除Index
      esClient.admin().indices().delete(new DeleteIndexRequest(indexName)).actionGet()
    }
    // 创建Index
    esClient.admin().indices().create(new CreateIndexRequest(indexName)).actionGet()
    // 声明写出时的ES配置信息
    val movieOptions = Map("es.nodes" -> esConf.httpHosts,
      "es.http.timeout" -> "100m",
      "es.mapping.id" -> "mid")
    // 电影数据写出时的Type名称【表】
    val movieTypeName = s"$indexName/$ES_MOVIE_TYPE"
    // 将Movie信息保存到ES
   /*  esMovieDF
       .write
       .options(movieOptions)
       .mode("overwrite")
       .format(ES_DRVIVE_CLASS)
       .save(movieTypeName)*/
    //这种方式不会报驱动问题
    EsSparkSQL.saveToEs(esMovieDF, movieTypeName)
  }


  def main(args: Array[String]): Unit = {

    //map封装配置信息
    val confMap = Map("spark.cores" -> "local[2]",
      "mongo.uri" -> MONGO_URI,
      "mongo.db" -> MONGODB_DATABASE,
      "es.httpHosts" -> ES_HTTPHOSTS,
      "es.transportHosts" -> ES_TRANSPORTHOSTS,
      "es.index" -> ES_INDEX,
      "es.cluster.name" -> ES_CLUSTER_NAME
    )

    // 定义MongoDB的配置对象
    implicit val mongoConf = MongoConfig(confMap("mongo.uri"), confMap("mongo.db"))

    // 定义ElasticSearch的配置对象
    implicit val esConf = ESConfig(confMap("es.httpHosts"), confMap("es.transportHosts"), confMap("es.index"), confMap("es.cluster.name"))

    // 声明Spark的配置信息
    val sparkConf = new SparkConf().setAppName("DataLoader").setMaster(confMap("spark.cores"))
    //增加es 的连接信息
    sparkConf.set("es.nodes", "192.168.187.130")
    sparkConf.set("es.port", "9200")
    sparkConf.set("es.mapping.id", "mid")
    sparkConf.set("es.write.operation", "index")
    // 创建SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // 加载Movie数据集
    val movieRDD = spark.sparkContext.textFile("D:\\6_ideaWorkPlace\\sxe_project01\\recommender\\dataload\\src\\main\\resources\\small\\movies.csv")

    // 加载Rating数据集
    val ratingRDD = spark.sparkContext.textFile("D:\\6_ideaWorkPlace\\sxe_project01\\recommender\\dataload\\src\\main\\resources\\small\\ratings.csv")

    // 加载Tag数据集
    val tagRDD = spark.sparkContext.textFile("D:\\6_ideaWorkPlace\\sxe_project01\\recommender\\dataload\\src\\main\\resources\\small\\tags.csv")

    //隐式转换
    import spark.implicits._

    // 将电影RDD转换为DataFrame
    val movieDF = movieRDD.map(line => {
      val x = line.split("\\^")
      //Movie(x(0).trim.toInt, x(1).trim, x(2).trim, x(3).trim, x(4).trim, x(5).trim, x(6).trim.split("\\|"), x(7).trim.split("\\|"), x(8).trim.split("\\|"), x(9).trim.split("\\|"))
      Movie(x(0).trim.toInt, x(1).trim, x(2).trim, x(3).trim, x(4).trim, x(5).trim, x(6).trim, x(7).trim, x(8).trim, x(9).trim)
    }).toDF()

    // 将评分RDD转换为DataFrame
    val ratingDF = ratingRDD.map(line => {
      val x = line.split(",")
      MoviesRating(x(0).toInt, x(1).toInt, x(2).toDouble, x(3).toInt)
    }).toDF()

    // 将标签RDD转换为DataFrame
    val tagDF = tagRDD.map(line => {
      val x = line.split(",")
      Tag(x(0).toInt, x(1).toInt, x(2).toString, x(3).toInt)
    }).toDF()
    //去除缓存
    tagDF.cache()
    movieDF.cache()
    // 将数据保存到MongoDB
    //  storeDataInMongo(movieDF, ratingDF, tagDF)


    //将tagDF中的标签合并在一起  爱情|动作....
    val tagCollectDF = tagDF.groupBy($"mid").agg(concat_ws("|", collect_set($"tag")).as("tags"))


    //将tags合并到movie数据集中产生新的movie数据集
    val esMovieDF = movieDF.join(tagCollectDF, Seq("mid", "mid"), "left").select("mid", "name", "descri", "timelong", "issue", "shoot", "language", "genres", "actors", "directors", "tags")

    // 将数据保存到ES
    storeDataInES(esMovieDF);

    //去除缓存
    tagDF.unpersist()
    movieDF.unpersist()


    spark.stop()

  }

}

/**
  * Movie Class 电影类
  *
  * @param mid       电影的ID
  * @param name      电影的名称
  * @param descri    电影的描述
  * @param timelong  电影的时长
  * @param issue     电影的发行时间
  * @param shoot     电影的拍摄时间
  * @param language  电影的语言
  * @param genres    电影的类别
  * @param actors    电影的演员
  * @param directors 电影的导演
  */
case class Movie(val mid: Int, val name: String, val descri: String, val timelong: String, val issue: String, val shoot: String, val language: String, val genres: String, val actors: String, val directors: String)

/**
  * Tag Class  电影标签类
  *
  * @param uid       用户的ID
  * @param mid       电影的ID
  * @param tag       用户为该电影打的标签
  * @param timestamp 用户为该电影打标签的时间
  */
case class Tag(val uid: Int, val mid: Int, val tag: String, val timestamp: Int)

/**
  * Rating Class 电影的评分类
  *
  * @param uid       用户的ID
  * @param mid       电影的ID
  * @param score     用户为该电影的评分
  * @param timestamp 用户为该电影评分的时间
  */
case class MoviesRating(val uid: Int, val mid: Int, val score: Double, val timestamp: Int)

/**
  * MongoDB 配置对象
  *
  * @param uri MongoDB连接地址
  * @param db  操作的MongoDB数据库
  */
case class MongoConfig(val uri: String, val db: String)

/**
  * ElasticSearch 配置对象
  *
  * @param httpHosts      ES的地址
  * @param transportHosts ES的通讯端口
  * @param index          ES操作的Index
  * @param clusterName    ES的集群名称
  */
case class ESConfig(val httpHosts: String, val transportHosts: String, val index: String, val clusterName: String)